"""
Case Type   : 基础功能
Case Name   : 修改wal_level
Create At   : 2022/1/29
Owner       : opentestcase026
Description :
    1.减容发布端为单节点
    2.修改wal_level为minimal
    3.创建发布端
    4.恢复发布端guc参数
    5.恢复发布端备节点
    6.修改wal_level为archive
    7.创建发布端
    8.修改wal_level为hot_standby
    9.创建发布端
Expect      :
    1.成功
    2.成功
    3.创建发布端成功，提示WARNING:  wal_level is insufficient to publish logical changes
    创建订阅端失败，提示ERROR:  Failed to connect to publisher
    4.成功
    5.成功
    6.成功
    7.创建发布端成功，提示WARNING:  wal_level is insufficient to publish logical changes
    创建订阅端失败，提示FATAL:  logical decoding requires wal_level >= logical
    8.成功
    9.创建发布端成功，提示WARNING:  wal_level is insufficient to publish logical changes
    创建订阅端失败，提示FATAL:  logical decoding requires wal_level >= logical
History     :
    Modified opentestcase026 代码变更，修改脚本适配
    Modified by opentestcase026 2022/5/11:修改wal_level设置方式
    Modified by opentestcase026 2022/5/23:修改执行互信节点
"""
import unittest
import os
import time
from yat.test import macro
from yat.test import Node
from testcase.utils.Logger import Logger
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Common import Common
from testcase.utils.Constant import Constant

Primary_SH = CommonSH('PrimaryDbUser')


@unittest.skipIf(3 != Primary_SH.get_node_num(), '非1+2环境不执行')
class Pubsubclass(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info("-----------this is setup-----------")
        self.log.info(f"-----{os.path.basename(__file__)[:-3]} start-----")
        self.pri_userdb_pub = Node(node='PrimaryDbUser')
        self.pri_root_pub = Node(node='PrimaryRoot')
        self.pri_userdb_sub = Node(node='remote1_PrimaryDbUser')
        self.constant = Constant()
        self.commsh_pub = CommonSH('PrimaryDbUser')
        self.commsh_pub_root = CommonSH('PrimaryRoot')
        self.commsh_pub_root_sta = CommonSH('Standby2Root')
        self.commsh_sub = CommonSH('remote1_PrimaryDbUser')
        self.com = Common()
        self.pubname = "pub_case046_1"
        self.subname = "sub_case046_1"
        self.parent_path_pub = os.path.dirname(macro.DB_INSTANCE_PATH)
        self.parent_path_sub = os.path.dirname(macro.DB_INSTANCE_PATH_REMOTE1)
        self.port = str(int(self.pri_userdb_pub.db_port) + 1)
        self.wal_level = self.com.show_param("wal_level")
        self.archive_mode = self.com.show_param("archive_mode")
        self.hot_standby = self.com.show_param("hot_standby")
        self.max_wal_senders = self.com.show_param("max_wal_senders")
        self.user_param_pub = f'-U {self.pri_userdb_pub.db_user} ' \
            f'-W {self.pri_userdb_pub.db_password}'
        self.user_param_sub = f'-U {self.pri_userdb_sub.db_user} ' \
            f'-W {self.pri_userdb_sub.db_password}'

        cmd = f"cp " \
            f"{os.path.join(macro.DB_INSTANCE_PATH, 'pg_hba.conf')} " \
            f"{os.path.join(self.parent_path_pub, 'pg_hba.conf')};"
        self.log.info(cmd)
        result = self.pri_userdb_pub.sh(f'source {macro.DB_ENV_PATH}; '
                                        f'gs_ssh -c "{cmd}"').result()
        self.log.info(result)
        cmd = f"cp " \
            f"{os.path.join(macro.DB_INSTANCE_PATH_REMOTE1, 'pg_hba.conf')}" \
            f" {os.path.join(self.parent_path_sub, 'pg_hba.conf')};"
        self.log.info(cmd)
        result = self.pri_userdb_sub.sh(cmd).result()
        self.log.info(result)
        cmd = f"cp " \
            f"{os.path.join(macro.DB_INSTANCE_PATH, 'postgresql.conf')} " \
            f"{os.path.join(self.parent_path_pub, 'postgresql.conf')};"
        result = self.pri_userdb_pub.sh(f'source {macro.DB_ENV_PATH}; '
                                        f'gs_ssh -c "{cmd}"').result()
        self.log.info(cmd)
        self.log.info(result)
        self.pub_stanode_num = self.commsh_pub.get_node_num() - 1
        self.log.info(self.pub_stanode_num)
        self.pub_sta_node = []
        sta_node_list = ['Standby1DbUser', 'Standby2DbUser']
        self.hosts = [self.pri_root_pub.ssh_host]
        self.comsh_sta_pub = []
        self.host_name_sta = []
        for i in range(self.pub_stanode_num):
            self.pub_sta_node.append(Node(node=sta_node_list[i]))
            self.hosts.append(self.pub_sta_node[i].ssh_host)
            self.comsh_sta_pub.append(CommonSH(sta_node_list[i]))
            self.host_name_sta.append(
                self.pub_sta_node[i].sh('hostname').result())
        self.params = {'-f': 'pubsub0046'}

    def set_pg_hba(self):
        text = '--修改pg_hba文件--'
        self.log.info(text)
        guc_res = self.commsh_pub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    replication  {self.pri_userdb_sub.db_user} '
            f'{self.pri_userdb_sub.db_host}/32 sha256')
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败:' + text)
        guc_res = self.commsh_pub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    all  {self.pri_userdb_pub.db_user} '
            f'{self.pri_userdb_pub.db_host}/32 sha256')
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败:' + text)
        guc_res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host    replication  {self.pri_userdb_pub.db_user} '
            f'{self.pri_userdb_pub.db_host}/32 sha256',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败:' + text)
        guc_res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host all  {self.pri_userdb_sub.db_user} '
            f'{self.pri_userdb_sub.db_host}/32 sha256',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败:' + text)

    def test_pubsub(self):
        try:
            text = '--step1:减容发布端为单节点 expect:成功--'
            self.log.info(text)
            for i in range(self.pub_stanode_num-1, -1, -1):
                if 0 == i:
                    lase_node = True
                else:
                    lase_node = False
                result = self.commsh_pub.exec_dropnode(
                    self.pub_sta_node[i], last=lase_node)
                self.assertTrue(result, '执行失败: 减容发布端为单节点')

            text = '--step2:修改wal_level为minimal expect:成功--'
            self.log.info(text)
            self.set_pg_hba()
            result = self.commsh_pub.execute_gsguc(
                'set ', self.constant.GSGUC_SUCCESS_MSG, 'wal_level=minimal')
            self.assertTrue(result, '执行失败:' + text)
            result = self.commsh_pub.execute_gsguc(
                'set ', self.constant.GSGUC_SUCCESS_MSG, 'archive_mode=off')
            self.assertTrue(result, '执行失败:' + text)
            result = self.commsh_pub.execute_gsguc(
                'set ', self.constant.GSGUC_SUCCESS_MSG, 'hot_standby=off')
            self.assertTrue(result, '执行失败:' + text)
            result = self.commsh_pub.execute_gsguc(
                'set ', self.constant.GSGUC_SUCCESS_MSG, 'max_wal_senders=0')
            self.assertTrue(result, '执行失败:' + text)
            result = self.commsh_pub.restart_db_cluster(True)
            flg = self.constant.START_SUCCESS_MSG in result \
                  or 'Degrade' in result
            self.assertTrue(flg, '执行失败:' + text)

            text = '--step3:创建发布端 expect:创建发布端成功，提示WARNING:  wal_level is ' \
                   'insufficient to publish logical changes创建订阅端失败，提示ERROR:  ' \
                   'Failed to connect to publisher--'
            self.log.info(text)
            sql = f"CREATE PUBLICATION {self.pubname} for all tables;" \
                f"select pg_sleep(10);" \
                f"select * from pg_PUBLICATION;"
            result = self.commsh_pub.execut_db_sql(
                sql, sql_type=self.user_param_pub)
            self.log.info(result)
            self.assertNotIn(self.constant.SQL_WRONG_MSG[1], result,
                             '执行失败:' + text)
            pub_warning_flg = f"WARNING:  wal_level is insufficient to " \
                f"publish logical changes"
            self.assertIn(pub_warning_flg, result, '执行失败:' + text)

            result = self.commsh_sub.execute_generate(
                macro.COMMON_PASSWD, env_path=macro.DB_ENV_PATH_REMOTE1)
            self.assertIn('', result, '执行失败:' + text)
            sql = f"CREATE SUBSCRIPTION {self.subname} CONNECTION " \
                f"'host={self.pri_userdb_pub.db_host} " \
                f"port={self.port} " \
                f"user={self.pri_userdb_pub.db_user} " \
                f"dbname={self.pri_userdb_pub.db_name} " \
                f"password={self.pri_userdb_pub.ssh_password}' " \
                f"PUBLICATION {self.pubname};"
            result = self.commsh_sub.execut_db_sql(sql, self.user_param_sub,
                                                   None,
                                                   macro.DB_ENV_PATH_REMOTE1)
            self.log.info(result)
            self.assertIn(self.constant.can_not_connect_pub_msg,
                          result, '执行失败:' + text)
        finally:
            text = '--step4:恢复发布端guc参数 expect:成功--'
            self.log.info(text)
            time.sleep(5)
            result = self.commsh_pub.execute_gsguc(
                'set ', self.constant.GSGUC_SUCCESS_MSG,
                f'wal_level={self.wal_level}')
            self.assertTrue(result, '执行失败:' + text)
            result = self.commsh_pub.restart_db_cluster(True)
            flg = self.constant.START_SUCCESS_MSG in result \
                  or 'Degrade' in result
            self.assertTrue(flg, '执行失败:' + text)
            result = self.commsh_pub.execute_gsguc(
                'set ', self.constant.GSGUC_SUCCESS_MSG,
                f'archive_mode={self.archive_mode}')
            self.assertTrue(result, '执行失败:' + text)
            result = self.commsh_pub.execute_gsguc(
                'set ', self.constant.GSGUC_SUCCESS_MSG,
                f'hot_standby={self.hot_standby}')
            self.assertTrue(result, '执行失败:' + text)
            result = self.commsh_pub.execute_gsguc(
                'set ', self.constant.GSGUC_SUCCESS_MSG,
                f'max_wal_senders={self.max_wal_senders}')
            self.assertTrue(result, '执行失败:' + text)
            result = self.commsh_pub.restart_db_cluster(True)
            flg = self.constant.START_SUCCESS_MSG in result \
                  or 'Degrade' in result
            self.assertTrue(flg, '执行失败:' + text)

            text = '--step5:恢复发布端备节点 expect:成功--'
            self.log.info(text)
            for loop in range(2):
                for i in range(self.pub_stanode_num):
                    time.sleep(2)
                    result = self.comsh_sta_pub[i].execute_gsguc(
                        'set', self.constant.GSGUC_SUCCESS_MSG,
                        'wal_level=hot_standby',
                        node_name=self.host_name_sta[i])
                    self.assertTrue(result, '执行失败:' + text)
                    result = self.comsh_sta_pub[i].execute_gsguc(
                        'set', self.constant.GSGUC_SUCCESS_MSG,
                        'hot_standby=off',
                        node_name=self.host_name_sta[i])
                    self.assertTrue(result, '执行失败:' + text)
                    result = self.comsh_sta_pub[i].execute_gsctl(
                        'restart', self.constant.REBUILD_SUCCESS_MSG)
                    self.assertTrue(result, '执行失败:' + text)
                    time.sleep(5)
                    result = self.comsh_sta_pub[i].execute_gsguc(
                        'set', self.constant.GSGUC_SUCCESS_MSG,
                        'hot_standby=on',
                        node_name=self.host_name_sta[i])
                    self.assertTrue(result, '执行失败:' + text)
                    result = self.comsh_sta_pub[i].execute_gsctl(
                        'restart', self.constant.REBUILD_SUCCESS_MSG)
                    self.assertTrue(result, '执行失败:' + text)
                    time.sleep(5)
                    result = self.comsh_sta_pub[i].stop_db_instance()
                    self.assertIn(self.constant.GS_CTL_STOP_SUCCESS_MSG,
                                  result, '执行失败:' + text)
                if 0 == loop:
                    ssh_res = self.commsh_pub_root_sta.exec_gs_sshexkey(
                        macro.DB_SCRIPT_PATH, *self.hosts, **self.params)
                    self.assertIn('Successfully created SSH trust.',
                                  ssh_res, '执行失败:发布端建立root互信')
                time.sleep(2)
                for i in range(self.pub_stanode_num):
                    result = self.commsh_pub_root.exec_expension(
                        self.pri_userdb_pub.ssh_user,
                        self.pri_userdb_pub.ssh_user,
                        self.pub_sta_node[i].ssh_host, macro.DB_XML_PATH)
                    if 1 == loop:
                        self.assertTrue(result, '执行失败: 扩容发布端备节点')
                    time.sleep(2)

        text = '--step6:修改wal_level为archive expect:成功--'
        self.log.info(text)
        self.set_pg_hba()
        result = self.commsh_pub.execute_gsguc(
            'set ', self.constant.GSGUC_SUCCESS_MSG,
            'hot_standby=off')
        self.assertTrue(result, '执行失败:' + text)
        result = self.commsh_pub.execute_gsguc(
            'set ', self.constant.GSGUC_SUCCESS_MSG, 'wal_level=archive')
        self.assertTrue(result, '执行失败:' + text)
        result = self.commsh_pub.restart_db_cluster(True)
        flg = self.constant.START_SUCCESS_MSG in result or 'Degrade' in result
        self.assertTrue(flg, '执行失败:' + text)

        text = '--step7:创建发布端 expect:创建发布端成功，' \
               '提示WARNING:  wal_level is insufficient to ' \
               'publish logical changes--'
        self.log.info(text)
        sql = f"drop PUBLICATION if exists {self.pubname};" \
            f"CREATE PUBLICATION {self.pubname} for all tables;"
        result = self.commsh_pub.execut_db_sql(
            sql, sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertNotIn(self.constant.SQL_WRONG_MSG[1], result,
                      '执行失败:' + text)
        self.assertIn(pub_warning_flg, result, '执行失败:' + text)
        sql = f"CREATE SUBSCRIPTION {self.subname} CONNECTION " \
            f"'host={self.pri_userdb_pub.db_host} " \
            f"port={self.port} " \
            f"user={self.pri_userdb_pub.db_user} " \
            f"dbname={self.pri_userdb_pub.db_name} " \
            f"password={self.pri_userdb_pub.ssh_password}' " \
            f"PUBLICATION {self.pubname};"
        result = self.commsh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        flg_pub = "FATAL:  logical decoding requires wal_level >= logical"
        self.assertIn(flg_pub, result, '执行失败:' + text)

        text = '--step8:修改wal_level为hot_standby expect:成功--'
        self.log.info(text)
        result = self.commsh_pub.execute_gsguc(
            'set ', self.constant.GSGUC_SUCCESS_MSG,
            'wal_level=hot_standby')
        self.assertTrue(result, '执行失败:' + text)
        result = self.commsh_pub.restart_db_cluster(True)
        flg = self.constant.START_SUCCESS_MSG in result or 'Degrade' in result
        self.assertTrue(flg, '执行失败:' + text)
        result = self.commsh_pub.execute_gsguc(
            'set ', self.constant.GSGUC_SUCCESS_MSG,
            'hot_standby=on')
        self.assertTrue(result, '执行失败:' + text)
        result = self.commsh_pub.restart_db_cluster(True)
        flg = self.constant.START_SUCCESS_MSG in result or 'Degrade' in result
        self.assertTrue(flg, '执行失败:' + text)

        text = '--step9:创建发布端 expect:创建发布端成功，' \
               '提示WARNING:  wal_level is insufficient to ' \
               'publish logical changes--'
        self.log.info(text)
        sql = f"drop PUBLICATION if exists {self.pubname};" \
            f"CREATE PUBLICATION {self.pubname} for all tables;"
        result = self.commsh_pub.execut_db_sql(
            sql, sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertNotIn(self.constant.SQL_WRONG_MSG[1], result,
                         '执行失败:' + text)
        self.assertIn(pub_warning_flg, result, '执行失败:' + text)
        sql = f"CREATE SUBSCRIPTION {self.subname} CONNECTION " \
            f"'host={self.pri_userdb_pub.db_host} " \
            f"port={self.port} " \
            f"user={self.pri_userdb_pub.db_user} " \
            f"dbname={self.pri_userdb_pub.db_name} " \
            f"password={self.pri_userdb_pub.ssh_password}' " \
            f"PUBLICATION {self.pubname};"
        result = self.commsh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn(flg_pub, result, '执行失败:' + text)

    def tearDown(self):
        self.log.info('------------this is tearDown-------------')
        text = '--清理环境--'
        self.log.info(text)
        result0 = self.commsh_pub.execute_gsguc(
            'set ', self.constant.GSGUC_SUCCESS_MSG,
            f'wal_level={self.wal_level}')
        self.commsh_pub.restart_db_cluster(True)
        result1 = self.commsh_pub.execute_gsguc(
            'set ', self.constant.GSGUC_SUCCESS_MSG,
            f'archive_mode={self.archive_mode}')
        result2 = self.commsh_pub.execute_gsguc(
            'set ', self.constant.GSGUC_SUCCESS_MSG,
            f'hot_standby={self.hot_standby}')
        result3 = self.commsh_pub.execute_gsguc(
            'set ', self.constant.GSGUC_SUCCESS_MSG,
            f'max_wal_senders={self.max_wal_senders}')
        self.commsh_pub.restart_db_cluster(True)
        cmd = f"mv " \
            f"{os.path.join(self.parent_path_pub, 'postgresql.conf')}  " \
            f"{os.path.join(macro.DB_INSTANCE_PATH, 'postgresql.conf')};"
        result = self.pri_userdb_pub.sh(f'source {macro.DB_ENV_PATH}; '
                                        f'gs_ssh -c "{cmd}"').result()
        self.log.info(result)
        sql = f"DROP PUBLICATION if exists {self.pubname};"
        drop_pub_result = self.commsh_pub.execut_db_sql(
            sql, sql_type=self.user_param_pub)
        self.log.info(drop_pub_result)
        sql = f"DROP SUBSCRIPTION if exists {self.subname};"
        drop_sub_result = self.commsh_sub.execut_db_sql(
            sql, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(drop_sub_result)
        cmd = f"mv " \
            f"{os.path.join(self.parent_path_pub, 'pg_hba.conf')} "\
            f"{os.path.join(macro.DB_INSTANCE_PATH, 'pg_hba.conf')} "
        self.log.info(cmd)
        result = self.pri_userdb_pub.sh(f'source {macro.DB_ENV_PATH}; '
                                        f'gs_ssh -c "{cmd}"').result()
        self.log.info(result)
        cmd = f"mv " \
            f"{os.path.join(self.parent_path_sub, 'pg_hba.conf')} "\
            f"{os.path.join(macro.DB_INSTANCE_PATH_REMOTE1, 'pg_hba.conf')} "
        self.log.info(cmd)
        result = self.pri_userdb_sub.sh(cmd).result()
        self.log.info(result)
        self.commsh_pub.restart_db_cluster(True)
        self.commsh_sub.restart_db_cluster(True, macro.DB_ENV_PATH_REMOTE1)
        self.assertIn(self.constant.drop_pub_succ_msg, drop_pub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.drop_sub_succ_msg, drop_sub_result,
                      '执行失败' + text)
        self.assertTrue(result2 and result1 and result0 and result3,
                        '执行失败:' + text)
        self.log.info(f"-----{os.path.basename(__file__)[:-3]} end-----")
